package cn.com.bluemoon.daps.sync.service.impl;

import cn.com.bluemoon.daps.common.domain.ResultBean;
import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.SyncStatus;
import cn.com.bluemoon.daps.common.enums.SyncToEnv;
import cn.com.bluemoon.daps.common.exception.DapThrowException;
import cn.com.bluemoon.daps.common.toolkit.BmAssetUtils;
import cn.com.bluemoon.daps.model.entity.DapDataModel;
import cn.com.bluemoon.daps.sync.base.BmScript;
import cn.com.bluemoon.daps.sync.base.CompareResult;
import cn.com.bluemoon.daps.sync.base.SqlCompareResult;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.ModifySyncStatusContext;
import cn.com.bluemoon.daps.sync.core.sync.TableModel;
import cn.com.bluemoon.daps.sync.core.sync.compare.ModelCompareClient;
import cn.com.bluemoon.daps.sync.entity.DapSyncProdEnvTask;
import cn.com.bluemoon.daps.sync.entity.DapSyncProdEnvTaskDetail;
import cn.com.bluemoon.daps.sync.feignclient.DapModelModuleFacade;
import cn.com.bluemoon.daps.sync.feignclient.DapSystemModuleFacade;
import cn.com.bluemoon.daps.sync.mapper.DapSyncProdEnvTaskMapper;
import cn.com.bluemoon.daps.sync.service.*;
import cn.com.bluemoon.daps.sync.vo.applypublish.ModelCompareVo;
import cn.com.bluemoon.daps.sync.vo.syncmanage.*;
import cn.com.bluemoon.daps.system.entity.DapSystemDatasourceEnvironment;
import cn.com.bluemoon.daps.system.entity.DapSystemInfo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.support.SFunction;
import com.google.common.collect.ImmutableList;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.map.CaseInsensitiveMap;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.util.function.Tuple2;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 同步管理
 * 1.获取所有模型的表同步信息
 * 2.针对同步失败的进行人工处理
 * 3.模型vs同步环境
 *
 * @author Jarod.Kong
 */
@Service
@Slf4j
public class DapSyncManagerServiceImpl implements DapSyncManagerService {
    public static final String DIFF_INFO_KEY = "diffInfo";
    /**
     * 变更状态处理
     */
    @Autowired
    private List<ModifySyncStatusHandler> modifySyncStatusHandlers;

    @Resource
    private DapSyncProdEnvTaskService syncProdEnvTaskService;
    @Resource
    private DapSyncProdEnvTaskDetailService syncProdEnvTaskDetailService;
    @Resource
    private DapModelModuleFacade modelModuleFacade;
    @Resource
    private DapSystemModuleFacade systemModuleFacade;
    @Resource
    private DapSyncBasicService syncBasicService;
    @Resource
    private DapSyncTaskOperRecordService taskOperRecordService;

    /**
     * 查询同步管理列表
     *
     * @param searchVo 查询条件
     * @return 分页
     */
    @SuppressWarnings("rawtypes")
    @Override
    public IPage<SyncManageOut> querySyncList(SyncManagementPageVo searchVo) {
        SyncManagementDto dto = new SyncManagementDto(searchVo);
        log.info("{}转换入参为：{}", searchVo, dto);
        List<DapModelModuleFacade.ModelInfo> modelInfos = modelModuleFacade.queryAllModels();
        Map<String, String> modelIdAndName = modelInfos.stream()
                .collect(Collectors.toMap(DapModelModuleFacade.ModelInfo::getModelId, DapModelModuleFacade.ModelInfo::getModelName, (a, b) -> b));
        if (StringUtils.isBlank(searchVo.getModelIdTrim())) {// 全部就取有效的模型ids
            dto.setModelIds(new ArrayList<>(modelIdAndName.keySet()));
        }
        IPage<SyncManageOut> res = syncProdEnvTaskService.getBaseMapper(DapSyncProdEnvTaskMapper.class)
                .selectSyncManagePage(searchVo.toMpPage(), dto)
                .convert((SFunction<SyncManageOut, SyncManageOut>) o -> {
                    String modelName = modelIdAndName.getOrDefault(o.getModelId(), null);
                    o.setDeletedModel(!modelIdAndName.containsKey(o.getModelId()));// 标记已被删
                    o.setModelName(modelName);
                    return o;
                });
        return res;
    }

    /**
     * 根据任务id+表id获取发布同步中的预发布+生产环境数据
     *
     * @param param a
     * @return 2 po
     */
    @Override
    public List<SyncEnvInfo> viewSyncEnvs(SyncTableParam param) {
        Optional<DapSyncProdEnvTaskDetail> detailOpt = syncProdEnvTaskDetailService.queryOneByParam(param);
        DapSyncProdEnvTaskDetail taskDetail = detailOpt.orElseThrow(() -> new DapThrowException("同步任务环境数据为空"));
        DapDataModel model = modelModuleFacade.getByIdAsset(param.getModelId(), "模型不存在！！");
        Optional<DapSystemInfo> systemInfo = systemModuleFacade.getSystemInfo(model.getSystemId());
        BmAssetUtils.isTrue(systemInfo.isPresent(), "获取系统信息失败！！");
        // 2021/6/16 根据syncToEnv数据源环境+数据源id获取源数据信息
        String dataSourceId = model.getDataSourceId();
        DapSyncProdEnvTask tasks = syncProdEnvTaskService.getByIdAsset(taskDetail.getTaskId(), "同步任务不存在！！");
        SyncEnvInfo syncEnvInfo = getSyncEnvInfo(tasks, taskDetail, dataSourceId, SyncToEnv.PRE_PROD, taskDetail.getPreProdScriptStatus());
        SyncEnvInfo syncEnvInfoP = getSyncEnvInfo(tasks, taskDetail, dataSourceId, SyncToEnv.PROD, taskDetail.getProdScriptStatus());
        merge(syncEnvInfo, syncEnvInfoP);
        return ImmutableList.of(syncEnvInfo, syncEnvInfoP);
    }

    @Override
    public List<DapModelModuleFacade.ModelInfo> modelInfo() {
        List<DapModelModuleFacade.ModelInfo> modelInfos = modelModuleFacade.queryAllModels();
        return modelInfos;
    }

    /**
     * 修复预发布为非同步，而生产为未同步的时候，同步时间获取错误
     *
     * @param syncEnvInfo  预发布
     * @param syncEnvInfoP 生产
     */
    private void merge(SyncEnvInfo syncEnvInfo, SyncEnvInfo syncEnvInfoP) {
        if (syncEnvInfo == null || syncEnvInfoP == null) {
            return;
        }
        if (SyncStatus.SYNC_WAIT.equals(syncEnvInfo.getSyncStatus())) {
            syncEnvInfo.setSyncTime(null);
        }
        if (SyncStatus.SYNC_WAIT.equals(syncEnvInfoP.getSyncStatus())) {
            syncEnvInfoP.setSyncTime(null);
        }
    }

    private SyncEnvInfo getSyncEnvInfo(DapSyncProdEnvTask tasks, DapSyncProdEnvTaskDetail taskDetail, String dataSourceId, SyncToEnv preProd, SyncStatus preProdScriptStatus) {
        GlobalConfig.DataSourceAndEnv dsDto = systemModuleFacade.getSysDataSourceAndEnvByIdAndEnv(dataSourceId, preProd);
        DapSystemDatasourceEnvironment env = dsDto.getEnv();
        return SyncEnvInfo.builder().modelId(taskDetail.getModelId())
                .taskDetailId(taskDetail.getId())
                .envCodeId(preProd.getCode())
                .host(env.getHost()).port(env.getPort()).schema(env.getDatabaseSchema()).dbname(env.getDatabaseName())
                .syncBy(taskDetail.getUpdateBy()).syncStatus(preProdScriptStatus)
                .dbaManageBy(taskDetail.getDbaManageBy())
                .syncTime(tasks.getSyncTime(preProd))
                .build();
    }

    @Override
    public String viewLogInfo(SyncTableEnvParam param) {
        SyncToEnv syncToEnv = getSyncToEnvPredicate(param);
        DapSyncProdEnvTaskDetail detailOpt = syncProdEnvTaskDetailService.getById(param.getTaskDetailId());
        return Optional.ofNullable(detailOpt).map(d -> d.getSyncLog(syncToEnv)).orElse("");
    }

    private SyncToEnv getSyncToEnvPredicate(SyncTableEnvParam param) {
        SyncToEnv syncToEnv = param.syncToEnvOpt().orElseThrow(() -> new DapThrowException("同步管理只允许预发布或生产环境处理"));
        BmAssetUtils.isTrue(syncToEnv.equals(SyncToEnv.PRE_PROD) || syncToEnv.equals(SyncToEnv.PROD), "同步管理只允许预发布或生产环境处理");
        return syncToEnv;
    }

    @Override
    public SyncConfirmOut confirmSync(SyncTableEnvParam param) {
        SyncToEnv syncToEnv = getSyncToEnvPredicate(param);
        Optional<DapSyncProdEnvTaskDetail> detailOpt = Optional.ofNullable(syncProdEnvTaskDetailService.getById(param.getTaskDetailId()));
        return detailOpt.map(d -> SyncConfirmOut.builder()
                .syncDdl(Optional.ofNullable(d.getDdl()).map(ddl->ddl.replace(BmScript.SQL_SPLIT_CHAR_RL, ";")).orElse(""))
                .realSyncDdl(d.getRealSyncDdl(syncToEnv))
                .dbTableSchema(d.getSyncPreTbSql(syncToEnv))
                .todo(SyncStatus.SYNC_ERROR.equals(d.getScriptSyncStatus(syncToEnv)))
                .build())
                .orElseThrow(() -> new DapThrowException("表同步结构信息为空"));
    }

    /**
     * 修改同步状态-前置：对比结果是否都符合强制更新状态sql
     *
     * @param param    入参
     * @param userInfo 用户
     * @return 结果，true修改成功，false失败+原因
     */
    @Override
    public boolean allMatchForceUpdateStatusPredicate(SyncTableEnvAndRealDdlParam param, UserInfoHolder.UserInfo userInfo) {
        SyncToEnv syncToEnv = getSyncToEnvPredicate(param);
        DapSyncProdEnvTaskDetail taskDetail = predicateTask(param, syncToEnv);
        // 模型信息校验和初始化全局配置
        List<ModelCompareVo> res = analyzeModelAndEnvDiff(param, userInfo, syncToEnv, Collections.singleton(taskDetail.getTableName()));
        List<ModelCompareVo> tableRes = res.stream().filter(r -> r.getTableName().equalsIgnoreCase(taskDetail.getTableName())).collect(Collectors.toList());
        // 所有脚本都符合改情况，就允许强制修改
        if (tableRes.isEmpty()) {
            return false;
        }
        boolean allMatchForceUpdateStatus = checkTableCompares(tableRes);
        return allMatchForceUpdateStatus;
    }

    @Override
    public ResultBean<String> syncStatus(SyncTableEnvAndRealDdlParam param, UserInfoHolder.UserInfo userInfo, SyncStatus syncStatus) {
        SyncToEnv syncToEnv = getSyncToEnvPredicate(param);
        DapSyncProdEnvTaskDetail taskDetail = predicateTask(param, syncToEnv);
        // 模型信息校验和初始化全局配置
        List<ModelCompareVo> res = analyzeModelAndEnvDiff(param, userInfo, syncToEnv, Collections.singleton(taskDetail.getTableName()));
        List<ModelCompareVo> tableRes = res.stream().filter(r -> r.getTableName().equalsIgnoreCase(taskDetail.getTableName())).collect(Collectors.toList());
        // 所有脚本都符合改情况，就允许强制修改
        boolean allMatchForceUpdateStatus = checkTableCompares(tableRes);
        if (!tableRes.isEmpty() && !allMatchForceUpdateStatus) {
            String error = String.format("发现待发布表结构与%s存在差异，请确保无差异再修改状态！！", syncToEnv.getDescEnv());
            List<String> sql = getDiffSqls(tableRes);
            return ResultBean.<String>error(error).addMoreData(DIFF_INFO_KEY, sql);
        }
        // 修改为完成的表状态数据+表的任务状态（在所有表都为已同步情况下，进行修改整个单号为完成）
        boolean updateStatusByParam = syncProdEnvTaskDetailService.updateStatusByParam(param, syncToEnv, userInfo);
        taskOperRecordService.mark(param.getModelId(), taskDetail.getTaskId(), String.format("同步管理-同步模型->%s，修改%s表更改状态为%s，执行人：%s",
                syncToEnv.getDescEnv(), taskDetail.getTableName(), updateStatusByParam ? "已同步" : "同步失败", userInfo.getUserName()));
        // 1.查看任务id下所有表都为已同步
        this.tryUpdateTaskSyncStatusBySyncEnv(param, syncToEnv, userInfo);
        return ResultBean.ok("修改状态完成");
    }

    @Override
    public ResultBean<String> syncRepairStatus(SyncTableEnvAndRealDdlParam param, UserInfoHolder.UserInfo userInfo, SyncStatus syncStatus) {
        SyncToEnv syncToEnv = getSyncToEnvPredicate(param);
        DapSyncProdEnvTaskDetail taskDetail = predicateTask(param, syncToEnv);
        // 模型信息校验和初始化全局配置
        analyzeDb2ModelDiff(param, userInfo, syncToEnv, Collections.singleton(taskDetail.getTableName()), true);
        // 再次确认是否完全覆盖
        List<ModelCompareVo> res = analyzeDb2ModelDiff(param, userInfo, syncToEnv, Collections.singleton(taskDetail.getTableName()), false);
        List<ModelCompareVo> tableRes = res.stream().filter(r -> r.getTableName().equalsIgnoreCase(taskDetail.getTableName())).collect(Collectors.toList());
        if (!tableRes.isEmpty()) {
            String error = String.format("发现待发布表结构与%s存在差异，请确保无差异再修改状态！！", syncToEnv.getDescEnv());
            return ResultBean.<String>error(error).addMoreData(DIFF_INFO_KEY, "");
        }
        // 修改为完成的表状态数据+表的任务状态（在所有表都为已同步情况下，进行修改整个单号为完成）
        boolean updateStatusByParam = syncProdEnvTaskDetailService.updateStatusByParam(param, syncToEnv, userInfo);
        taskOperRecordService.mark(param.getModelId(), taskDetail.getTaskId(), String.format("同步管理-同步%s->模型，修改%s表更改状态为%s，执行人：%s",
                syncToEnv.getDescEnv(), taskDetail.getTableName(), updateStatusByParam ? "已同步" : "同步失败", userInfo.getUserName()));
        // 1.查看任务id下所有表都为已同步
        this.tryUpdateTaskSyncStatusBySyncEnv(param, syncToEnv, userInfo);
        return ResultBean.ok("修改状态完成");
    }
    /**
     * 判断表对比结果是否都符合“可强制更改状态”
     *
     * @param compareVos 对比数据
     * @return true可以 false不可
     */
    private boolean checkTableCompares(List<ModelCompareVo> compareVos) {
        // 若存在非字段修改的语句，直接结束false
        boolean allMatchForceUpdateStatus = compareVos.stream().allMatch(compareVo -> {
            // 2021/10/27 改为拦截/过滤-职责链模式进行match，只要符合一种就认为可以force
            ModifySyncStatusContext syncStatusContext = new ModifySyncStatusContext(modifySyncStatusHandlers, compareVo);
            return syncStatusContext.anyMatch(compareVo);
        });
        log.warn("分析表对比脚本是否可强制更改状态，结果：{}", allMatchForceUpdateStatus);
        return allMatchForceUpdateStatus;
    }

    private List<String> getDiffSqls(List<ModelCompareVo> res) {
        List<String> sql = new ArrayList<>();
        for (ModelCompareVo re : res) {
            sqlCollector(sql, re.getTableCompares());
            sqlCollector(sql, re.getFieldCompares());
            sqlCollector(sql, re.getIndexCompares());
            sqlCollector(sql, re.getPartitionCompares());
        }
        return sql;
    }

    private void sqlCollector(List<String> sql, List<CompareResult<SqlCompareResult>> compareResults) {
        if (compareResults != null && !compareResults.isEmpty()) {
            sql.addAll(compareResults.stream().map(s -> s.getData().getScript().getSql()).collect(Collectors.toList()));
        }
    }

    /**
     * 尝试更新整个任务单号状态
     *
     * @param param     入参
     * @param syncToEnv 环境
     */
    private void tryUpdateTaskSyncStatusBySyncEnv(SyncTableEnvParam param, SyncToEnv syncToEnv, UserInfoHolder.UserInfo userInfo) {
        DapSyncProdEnvTaskDetail taskDetail = syncProdEnvTaskDetailService.getById(param.getTaskDetailId());
        LambdaQueryWrapper<DapSyncProdEnvTaskDetail> queryNotSyncDoneWhere = syncProdEnvTaskDetailService.bmLambdaQuery()
                .eq(DapSyncProdEnvTaskDetail::getTaskId, taskDetail.getTaskId())
                .ne(SyncToEnv.PRE_PROD.equals(syncToEnv), DapSyncProdEnvTaskDetail::getPreProdScriptStatus, SyncStatus.SYNC_DONE)
                .ne(SyncToEnv.PROD.equals(syncToEnv), DapSyncProdEnvTaskDetail::getProdScriptStatus, SyncStatus.SYNC_DONE);
        int notSyncDoneSize = syncProdEnvTaskDetailService.count(queryNotSyncDoneWhere);
        boolean allDone = notSyncDoneSize == 0;
        if (allDone) {
            LambdaUpdateWrapper<DapSyncProdEnvTask> updateSet = syncProdEnvTaskService.bmLambdaUpdate().eq(DapSyncProdEnvTask::getModelId, param.getModelId())
                    .eq(DapSyncProdEnvTask::getId, taskDetail.getTaskId())
                    .set(DapSyncProdEnvTask::getSandboxSyncFail, Boolean.FALSE)// 更改状态成功后把执行在沙箱失败的置为false，为了发布同步查看脚本能显示为已执行
                    .set(SyncToEnv.PRE_PROD.equals(syncToEnv), DapSyncProdEnvTask::getPreProdStatus, SyncStatus.SYNC_DONE)
                    .set(SyncToEnv.PROD.equals(syncToEnv), DapSyncProdEnvTask::getProdStatus, SyncStatus.SYNC_DONE);
            boolean done = syncProdEnvTaskService.update(updateSet);
            taskOperRecordService.mark(param.getModelId(), taskDetail.getTaskId(), String.format("同步管理-同步%s，同步任务完成，执行人：%s",
                    syncToEnv.getDescEnv(), userInfo.getUserName()));
            log.info("同步管理，修改表状态且处理完成，该模型单号为完成，status:{}", done);
            // 增加消息通知
            DapSyncProdEnvTask task = syncProdEnvTaskService.getById(taskDetail.getTaskId());
            GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(task.getModelId(), syncToEnv, userInfo);
            log.info("发送同步管理修改任务完成消息，taskId={}", task.getId());
            syncBasicService.sendEmailPublishSyncFromSyncManagement(task, globalConfig);
        }
    }

    /**
     * 分析模型与同步环境的差异数据
     *
     * @param param     入参
     * @param userInfo  用户信息
     * @param syncToEnv 同步到
     * @return 对比结果
     */
    private List<ModelCompareVo> analyzeModelAndEnvDiff(SyncTableEnvParam param, UserInfoHolder.UserInfo userInfo, SyncToEnv syncToEnv, Set<String> tableNameIds) {
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(param.getModelId(), syncToEnv, userInfo);
        globalConfig.setModelNewest(false);
        globalConfig.setCanDropDbTable(true);
        Tuple2<List<TableModel>, List<TableModel>> localAndMetadataTableModel = syncBasicService.getLocalAndMetadataTableModel(globalConfig);
        CaseInsensitiveMap<String, String> caseInsensitiveTableNames = new CaseInsensitiveMap<>(tableNameIds.size() * 2);
        for (String t : tableNameIds) {
            caseInsensitiveTableNames.put(t, t);
        }
        List<TableModel> modelV1 = localAndMetadataTableModel.getT1().stream().filter(m -> caseInsensitiveTableNames.containsKey(m.getTable().getEngName())).collect(Collectors.toList());
        List<TableModel> modelV2 = localAndMetadataTableModel.getT2().stream().filter(m -> caseInsensitiveTableNames.containsKey(m.getTable().getEngName())).collect(Collectors.toList());
        List<ModelCompareVo> res = ModelCompareClient.doCompareV2(ModelCompareClient.BizType.SQL, globalConfig, modelV2, modelV1);
        log.info("修改状态前，对比结果为{}", res);
        return res;
    }

    /**
     * 分析模型与同步环境的差异数据
     *
     * @param param     入参
     * @param userInfo  用户信息
     * @param syncToEnv 同步到
     * @return 对比结果
     */
    private List<ModelCompareVo> analyzeDb2ModelDiff(SyncTableEnvParam param, UserInfoHolder.UserInfo userInfo,
                                                     SyncToEnv syncToEnv, Set<String> tableNameIds,
                                                     boolean addSinkDiff) {
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(param.getModelId(), syncToEnv, userInfo);
        globalConfig.setModelNewest(false);
        globalConfig.setCanDropDbTable(true);
        globalConfig.setAddSink(addSinkDiff);
        Tuple2<List<TableModel>, List<TableModel>> localAndMetadataTableModel = syncBasicService.getLocalAndMetadataTableModel(globalConfig);
        CaseInsensitiveMap<String, String> caseInsensitiveTableNames = new CaseInsensitiveMap<>(tableNameIds.size() * 2);
        for (String t : tableNameIds) {
            caseInsensitiveTableNames.put(t, t);
        }
        List<TableModel> modelV1 = localAndMetadataTableModel.getT1().stream().filter(m -> caseInsensitiveTableNames.containsKey(m.getTable().getEngName())).collect(Collectors.toList());
        List<TableModel> modelV2 = localAndMetadataTableModel.getT2().stream().filter(m -> caseInsensitiveTableNames.containsKey(m.getTable().getEngName())).collect(Collectors.toList());
        // 生产环境V1 to 模型V2
        List<ModelCompareVo> res = ModelCompareClient.doCompareV2(ModelCompareClient.BizType.Db2Model, globalConfig,modelV1, modelV2);
        log.info("修改状态前，对比结果为{}", res);
        return res;
    }

    private DapSyncProdEnvTaskDetail predicateTask(SyncTableEnvParam param, SyncToEnv syncToEnv) {
        // 判断是否该任务为同步失败
        DapSyncProdEnvTaskDetail taskDetail = syncProdEnvTaskDetailService.getByIdAsset(param.getTaskDetailId(), "任务表同步环境数据为空");
        BmAssetUtils.isTrue(SyncStatus.SYNC_ERROR.equals(taskDetail.getScriptSyncStatus(syncToEnv)), "只允许任务状态为同步失败");
        return taskDetail;
    }
}
